package com.example.demo.kafka;

import com.alibaba.fastjson.JSON;
import com.example.demo.yzy.server.impl.TestSeverImpl;
import com.example.demo.yzy.vo.UpdateVo;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * @author: 20971
 * @time: 2022/2/17 10:17
 * @description:
 */
@Component
public class KafkaConsumer {

    private static Logger log = LoggerFactory.getLogger(TestSeverImpl.class);

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    // 新建一个异常处理器，用@Bean注入
    @Bean
    public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
        return (message, exception, consumer) -> {
            log.error(exception.getCause()+"消费信息："+message.getPayload());
            log.error("异常:"+exception.getMessage());
            return message;
        };
    }
    @KafkaListener(id = "consumer0",topics = {"kafka_test"},groupId = "test-group",errorHandler = "consumerAwareErrorHandler")
    public void listener(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        log.info("消费："+record.topic()+"-"+record.partition());
        String key = (String) record.key();
        String value =  (String)record.value();
        if (StringUtils.isNoneBlank(key) && StringUtils.isNoneBlank(value)) {
            UpdateVo vo = JSON.parseObject(value, UpdateVo.class);
            //防止重复消费 调用SETNX操作获取锁，如果返回true ，代表获取锁成功
            Boolean aBoolean = redisTemplate.opsForValue().setIfAbsent(JSON.toJSONString(vo), null, 10, TimeUnit.HOURS);
            log.info("aBoolean:"+aBoolean);
            if (aBoolean != null && aBoolean) {
                log.info("消费端key:"+key);
                log.info("消费端value:"+value);
            }
            //防止消息丢失
            redisTemplate.delete(String.valueOf(vo.getId()));
        }
        //手动提交
        ack.acknowledge();
    }
}
